package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.DimUtil;
import com.atguigu.edu.realtime.util.HbaseUtil;
import com.atguigu.edu.realtime.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;


/**
 * ClassName: DimMapFunction
 * Package: com.atguigu.gmall.realtime.app.func
 * Description:
 * 涉及知识点:
 *    1. *** java设计模式 - 模板方法设计模式
 *       描述: 在父类中定义完成某一个功能的核心算法的骨架(步骤), 但是，具体的某些实现步骤 延迟到子类中去实现(抽象),
 *         在不改变父类核心(算法)骨架的前提下，每一个子类都可以有自己不同的实现。
 * 实现需求: 维度关联的基本实现
 * 1.
 *    public abstract class RichMapFunction<IN, OUT> extends AbstractRichFunction
 *         implements MapFunction<IN, OUT>
 *      <IN> Type of the input elements.
 *      <OUT> Type of the returned elements.
 * 2.
 *     public abstract OUT map(IN value) throws Exception;
 * @Author Mr.2
 * @Create 2023/9/2 9:04
 * @Version 1.0
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T>{
    /*
    * 使用泛型模板 定义任意数据类型
    * <IN>  - T
    * <OUT> - T
    *
    * */

    // 声明 Hbase 连接;
    private Connection connection;
    // 声明 RedisClient 连接;
    private Jedis jedis;
    //
    private String tableName;

    //
    public DimMapFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建 Hbase 连接
        connection = HbaseUtil.getHbaseConnect();
        // 创建 Jedis(就是RedisClient)
        jedis = RedisUtil.getJedis();
    }

    // 流处理中, close()方法不会被调用; 只有在处理完数据的时候, 才会被调用
    @Override
    public void close() throws Exception {
        // Note that: 若, 在流数据处理过程中，close()方法被调用，程序终止;
        HbaseUtil.closeHbaseConnect(connection);
        // 关闭 Jedis
        RedisUtil.closeJedis(jedis);
    }


    @Override
    public T map(T object) throws Exception {
        // 1. 根据 流中的对象，获取 要关联的维度主键
        String redisRowKey = getKey(object);

        // 2. 根据 维度的主键，获取 维度对象
        JSONObject dimInfoJsonObj =
                DimUtil.getDimInfo(jedis, connection, EduConfig.HBASE_NAMESPACE, tableName, redisRowKey);


        if (dimInfoJsonObj != null) {
            // 3. 将维度对象的属性补充到 数据流中的对象上
            join(object, dimInfoJsonObj);
        }

        return object;
    }

}
